package com.gold.zookeeperdemo.lock;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 *      zk分布式锁-临时顺序节点实现版
 *          推荐使用：避免死锁；避免惊群效应
 *      @author yf
 */
public class ZkDistributedLock implements Lock{
    /** 创建一个zk客户端 */
    private ZkClient zkClient=new ZkClient("47.97.23.184:2181");

    {
        zkClient.setZkSerializer(new MyZkSerializer());
    }
    private static final String LOCK_PATH="/LOCK";
    /** 当前的节点路径*/
    private ThreadLocal<String> currentPath=new ThreadLocal<>();
    /** 前一位的节点路径*/
    private ThreadLocal<String> beforePath=new ThreadLocal<>();

    public ZkDistributedLock() {
        super();
        if(!zkClient.exists(LOCK_PATH)){
            zkClient.createPersistent(LOCK_PATH);
        }
    }

    /**
     * 1.给自己注册一个临时顺序节点； 2. 获取所有子节点，判断自己的节点是否是最小的
     * 3.若是最小的，则获得锁；否则阻塞自己，注册对前一个节点的watcher，等待锁。
     * @return
     */
    @Override
    public boolean tryLock() {
        //防止重复创建临时顺序节点
        if(currentPath.get()==null){
            currentPath.set(
                    //为自己创建一个临时顺序节点
                    zkClient.createEphemeralSequential(LOCK_PATH + "/","lockTmp")
            );
        }
        //获取所有子节点列表
        List<String> children=zkClient.getChildren(LOCK_PATH);
        //排序
        Collections.sort(children);
        //获取最小子节点
        String minChildPath=LOCK_PATH+"/"+children.get(0);
        //判断当前节点是否是最小子节点：若是，则获得锁
        if(currentPath.get().equals(minChildPath)){
            return true;
        }
        //获取当前子节点，在List中的下标index
        //Collections.binarySearch(list,T key)使用二叉搜索算法搜索指定对象的指定列表
        int currentIndex= Collections.binarySearch(children,
                //substring(x)是从字符串的的第x个字符截取，范围 (x,length)
                currentPath.get().substring(LOCK_PATH.length()+1));
        //得到前一个子节点
        beforePath.set(LOCK_PATH + "/" + children.get(currentIndex-1));

        return false;
    }

    @Override
    public void lock() {
        if(!tryLock()){
            waitForLock();
            lock();
        }
    }

    /**
     *  等待锁-阻塞逻辑
     */
    private void waitForLock(){
        //阻塞自己
        CountDownLatch latch=new CountDownLatch(1);
        //注册对前一个节点的watcher
        IZkDataListener listener=new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {

            }
            @Override
            public void handleDataDeleted(String s) throws Exception {
                //收到节点删除通知，唤醒等待的线程
                latch.countDown();
                System.out.println("收到前一个节点删除通知,当前删除节点 path="+s);
            }
        };
        //订阅前一节点的监听
        zkClient.subscribeDataChanges(beforePath.get(),listener);

        //若前一个节点存在，则阻塞，等待节点变化监听-唤醒
        if(zkClient.exists(beforePath.get())){
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //移除订阅
        zkClient.unsubscribeDataChanges(beforePath.get(),listener);

    }

    @Override
    public void unlock() {
        zkClient.delete(currentPath.get());
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }



    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }



    @Override
    public Condition newCondition() {
        return null;
    }
}
